package com.hisense.ovcloud.common3.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.stereotype.Component;
import reactor.core.scheduler.Scheduler;

@Component
@Slf4j
@Lazy
@ConditionalOnClass(KafkaAutoConfiguration.class)
public class ReactiveKafkaSender {

    @Value("${printRedisTime:false}")
    private boolean printRedisTime;
    private final ReactiveKafkaProducerTemplate<String, String> kafkaProducerTemplate;

    public ReactiveKafkaSender(ReactiveKafkaProducerTemplate<String, String> kafkaProducerTemplate) {
        this.kafkaProducerTemplate = kafkaProducerTemplate;
    }

    public void sendToKafkaReactiveShard(String topic, String key, String data, Scheduler scheduler) {
        if(printRedisTime){
            kafkaProducerTemplate.send(topic, key, data)
                .doOnSuccess(senderResult -> log.info("sent to topic: {} -> {} ,offset: {}, partition: {}", topic, data, senderResult.recordMetadata().offset(), senderResult.recordMetadata().partition()))
                    .subscribeOn(scheduler)
                    .subscribe();
        }else{
            kafkaProducerTemplate.send(topic, key, data)
                    .subscribeOn(scheduler)
                    .subscribe();
        }

    }

    public void sendToKafkaReactiveDirect(String topic, String data, Scheduler scheduler) {
        if(printRedisTime){
            kafkaProducerTemplate.send(topic, data)
                    .doOnSuccess(senderResult -> log.info("sent to topic: {} -> {} ,offset: {}, partition: {}", topic, data, senderResult.recordMetadata().offset(), senderResult.recordMetadata().partition()))
                    .subscribeOn(scheduler)
                    .subscribe();
        }else{
            kafkaProducerTemplate.send(topic, data)
                    .subscribeOn(scheduler)
                    .subscribe();
        }
    }
}
